package com.swak.paxos.impl;

import com.swak.paxos.Acceptor;
import com.swak.paxos.PaxosConfig;
import com.swak.paxos.breakpoint.AcceptorBP;
import com.swak.paxos.enums.ProposalType;
import com.swak.paxos.protocol.PaxosMessage;
import com.swak.paxos.state.AcceptorState;
import com.swak.paxos.state.BallotNumber;
import com.swak.paxos.store.PaxosLog;
import com.swak.paxos.store.PaxosStorage;

import lombok.Getter;

/**
 * PaxosAcceptor.
 * 
 * @author DELL
 */
@Getter
public class PaxosAcceptor implements Acceptor {

	private final PaxosNode node;
	private final PaxosConfig config;
	private final PaxosGroup group;
	private final PaxosInstance instance;
	private final PaxosStorage storage;

	private final AcceptorState state = new AcceptorState();

	public PaxosAcceptor(PaxosConfig config, PaxosNode node, PaxosGroup group, PaxosInstance instance,
			PaxosStorage storage) {
		this.config = config;
		this.node = node;
		this.group = group;
		this.instance = instance;
		this.storage = storage;
	}

	@Override
	public void newInstance() {
		this.state.init();
	}

	@Override
	public void onPrepare(PaxosMessage message) {

		AcceptorBP.onPrepare(this.group.getGroupID(), this.instance.getInstanceID(), message.getProposalID());

		// 创建响应
		PaxosMessage reply = new PaxosMessage();
		reply.setInstanceID(this.instance.getInstanceID());
		reply.setNodeID(this.node.id());
		reply.setProposalID(message.getProposalID());
		reply.setType(ProposalType.PrepareReply);

		// 此轮投票
		BallotNumber ballot = new BallotNumber(message.getProposalID(), message.getNodeID());

		// 之前 Promise 的投票
		BallotNumber pbn = this.state.getPromiseBallot();

		// 此轮投票 必须大于 之前Promise的投票
		if (ballot.gt(pbn)) {

			// 之前接收阶段接收过
			BallotNumber acceptedBallot = this.state.getAcceptedBallot();

			// 返回之前接收阶段接收过的一些值
			reply.setPreAcceptID(acceptedBallot.getProposalID());
			reply.setPreAcceptNodeID(acceptedBallot.getNodeID());

			// 这个值很关键，就是确定的这个值(这个值是多数派确定过的值)
			if (acceptedBallot.getProposalID() > 0) {
				reply.setValue(this.state.getAcceptedValue());
			}

			// Promise ballot
			this.state.setPromiseBallot(ballot);

			// 记录日志
			// 如果不回复则会导致超时处理
			// 存储是最基本的
			boolean success = this.updateAcceptorState4Prepare(ballot);
			if (!success) {
				return;
			}
		}

		// 否则拒绝
		else {
			AcceptorBP.onPrepareReject(this.group.getGroupID(), this.instance.getInstanceID(), pbn.getProposalID());
			reply.setRejectByPromiseID(this.state.getPromiseBallot().getProposalID());
		}

		// 回复消息 -- 回复到指定的 node.
		this.sendMessage(message.getNodeID(), reply);
	}

	private boolean updateAcceptorState4Prepare(BallotNumber ballot) {
		boolean success = this.persist(this.instance.getInstanceID());
		if (success) {
			AcceptorBP.onPreparePass(this.group.getGroupID(), this.instance.getInstanceID(), ballot.getProposalID());
			return success;
		}
		AcceptorBP.onPreparePersistFail(this.group.getGroupID(), this.instance.getInstanceID(), ballot.getProposalID());
		return success;
	}

	private boolean persist(long instanceID) {

		// 投票的数据
		BallotNumber promiseBallot = this.state.getPromiseBallot();
		BallotNumber acceptedBallot = this.state.getAcceptedBallot();

		// 持久化数据
		PaxosLog log = new PaxosLog();
		log.setInstanceID(instanceID);
		log.setPromiseID(promiseBallot.getProposalID());
		log.setPromiseNodeID(promiseBallot.getNodeID());
		log.setAcceptedID(acceptedBallot.getProposalID());
		log.setAcceptedNodeID(acceptedBallot.getNodeID());
		log.setAcceptedValue(this.state.getAcceptedValue());
		return this.storage.storage(log);
	}

	@Override
	public void onAccept(PaxosMessage message) {

		AcceptorBP.onAccept(this.group.getGroupID(), this.instance.getInstanceID(), message.getProposalID());

		// 创建响应
		PaxosMessage reply = new PaxosMessage();
		reply.setInstanceID(this.instance.getInstanceID());
		reply.setNodeID(this.node.id());
		reply.setProposalID(message.getProposalID());
		reply.setType(ProposalType.AcceptReply);

		// 此轮投票
		BallotNumber ballot = new BallotNumber(message.getProposalID(), message.getNodeID());

		// 之前 Promise 的投票
		BallotNumber pbn = this.state.getPromiseBallot();

		// 此轮投票 必须大于 之前Promise的投票
		if (ballot.ge(pbn)) {
			BallotNumber acceptedBallot = new BallotNumber(ballot.getProposalID(), ballot.getNodeID());
			this.state.setAcceptedBallot(acceptedBallot);
			this.state.setAcceptedValue(message.getValue());
			this.state.setPromiseBallot(ballot);
			this.updateAcceptorState4Accept(ballot);
		}

		// 否则拒绝
		else {
			AcceptorBP.onAcceptReject(this.group.getGroupID(), this.instance.getInstanceID(), pbn.getProposalID());
			reply.setRejectByPromiseID(this.state.getPromiseBallot().getProposalID());
		}

		// 发送消息
		this.sendMessage(message.getNodeID(), reply);
	}

	void updateAcceptorState4Accept(BallotNumber ballot) {
		boolean success = this.persist(this.instance.getInstanceID());
		if (success) {
			AcceptorBP.onAcceptPass(this.group.getGroupID(), this.instance.getInstanceID(), ballot.getProposalID());
			return;
		}
		AcceptorBP.onAcceptPersistFail(this.group.getGroupID(), this.instance.getInstanceID(), ballot.getProposalID());
	}

	void sendMessage(String nodeID, PaxosMessage message) {

		// 当前节点不需要经过网络
		if (this.node.id().equals(nodeID)) {
			this.instance.doInnerReceivePaxosMessage(message);
			return;
		}

		// 通过网络发送
		this.node.sendMessage(nodeID, message);
	}

	@Override
	public boolean checkAcceptBallot(PaxosMessage message) {

		/*
		 * 校验 InstanceID
		 * InstanceID 不同： 不能学习
		 */
		if (message.getInstanceID() != this.instance.getInstanceID()) {
			return false;
		}

		/*
		 * 此轮没有接受过提案
		 * 必须接受过提案
		 */
		if (this.state.getAcceptedBallot().getProposalID() == 0) {
			return false;
		}

		/*
		 * 此轮投票必须和接受的投票一样： 完全一致
		 */
		BallotNumber ballot = new BallotNumber(message.getProposalID(), message.getNodeID());
		BallotNumber thisBallot = this.state.getAcceptedBallot();
		if (thisBallot.getProposalID() != ballot.getProposalID() // ProposalID
				|| !thisBallot.getNodeID().equals(ballot.getNodeID())) { // NodeID
			return false;
		}
		return true;
	}
}
